package com.cmb.lk50.sink.es;

import com.cmb.lk50.pojo.AbstractLog;
import com.cmb.lk50.sink.AsyncLogSink;
import io.searchbox.client.JestClient;
import io.searchbox.core.DocumentResult;
import io.searchbox.core.Index;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Base64;
import java.util.concurrent.ExecutorService;

/**
 * ES sink
 * Author       : 蒋俊钊 80256973
 * Date         : 2019/11/26
 * Copyright    (C) ChinaMerchantsBank
 */
public class EsSink extends AsyncLogSink {

    private static final Logger LOGGER = LoggerFactory.getLogger(EsSink.class);

    private JestClient jestClient;

    private String indexName;

    private String typeName;

    private String auth;

    public EsSink(JestClient jestClient, String indexName, String typeName, String username, String passwd, ExecutorService asyncExecutor) {
        super(asyncExecutor);
        this.jestClient = jestClient;
        this.indexName = indexName;
        this.typeName = typeName;
        this.auth = "Basic " + Base64.getEncoder().encodeToString((username + ":" + passwd).getBytes());
    }

    @Override
    protected void asyncWrite(AbstractLog log) {
        Index index = new Index.Builder(log)
          .setHeader("Authorization", this.auth)
          .index(log.esIndex() == null ? indexName : log.esIndex())
          .type(log.esType() == null ? typeName : log.esType())
          .build();

        DocumentResult result = null;
        try {
            result = jestClient.execute(index);
        } catch (IOException e) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("日志写入ES异常, [{}]", e.getMessage());
            }
        }

        if (LOGGER.isDebugEnabled()) {
            if (result == null) {
                LOGGER.debug("日志写入ES, 返回结果为空");
            } else if (!result.isSucceeded()) {
                LOGGER.debug("日志写入ES失败, [{}]", result.getErrorMessage());
            }
        }
    }
}